package com.itheima.kafkastreamdemo.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 承上启下
 * 从itheima-topic-stream-input接受消息
 * 每10秒聚合一次
 * 发送到下游itheima-topic-stream-out
 */
public class KafkaStreamWordCountDemo {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092"); //kafka地址
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());//key的序列化类
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());//value的序列化类
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"kafka-stream-id");


        StreamsBuilder streamsBuilder = new StreamsBuilder();

        //流式计算业务
        streamPrcesssor(streamsBuilder);

        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build() ,properties);

        kafkaStreams.start();//开启流式计算
    }

    /**
     * 流式计算
     * @param streamsBuilder
     *
     * hello 2
     * kafka 1
     * itcast 1
     */
    private static void streamPrcesssor(StreamsBuilder streamsBuilder) {
        //监听itheima-topic-stream-input  主题
        KStream<String, String> stream = streamsBuilder.stream("itheima-topic-stream-input");

        //流式处理
        stream.flatMapValues( value -> Arrays.asList(value.split(" ")) )//按消息  空格拆分得到所有的单词
        .groupBy((key,value)-> value)//按value分组
        .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//每10秒聚合一次
        .count()//统计数量
        .toStream()//转换为Kstream对象
        .map((key, value) -> {
            System.out.println("key:"+key+"   value:"+value);
            return new KeyValue<>(key.key().toString(), value.toString());})// key:hello  value:2
        .to("itheima-topic-stream-out");
    }

}
